package io.kiki.sba.registry.server.clientInterface.registry;


import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.api.exchange.RequestChannelClosedException;
import io.kiki.sba.registry.common.model.ClientToClientInterfaceServerNodeId;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.store.*;
import io.kiki.sba.registry.common.model.wrapper.Wrapper;
import io.kiki.sba.registry.common.model.wrapper.WrapperInvocation;
import io.kiki.sba.registry.server.clientInterface.acceptor.ClientOffWriteDataRequest;
import io.kiki.sba.registry.server.clientInterface.acceptor.PublisherWriteDataRequest;
import io.kiki.sba.registry.server.clientInterface.acceptor.WriteDataAcceptor;
import io.kiki.sba.registry.server.clientInterface.acceptor.WriteDataRequest;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.metadata.MetadataCacheRegistry;
import io.kiki.sba.registry.server.clientInterface.providedata.ConfigProvideDataWatcher;
import io.kiki.sba.registry.server.clientInterface.push.FirePushService;
import io.kiki.sba.registry.server.clientInterface.push.PushSwitchService;
import io.kiki.sba.registry.server.clientInterface.push.TriggerPushContext;
import io.kiki.sba.registry.server.clientInterface.store.Publishers;
import io.kiki.sba.registry.server.clientInterface.store.Subscribers;
import io.kiki.sba.registry.server.clientInterface.store.Watchers;
import io.kiki.sba.registry.server.clientInterface.strategy.RegistryStrategy;
import io.kiki.sba.registry.server.clientInterface.wrapper.RegisterInvokeData;
import io.kiki.sba.registry.server.clientInterface.wrapper.WrapperInterceptorManager;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;


public class ClientInterfaceRegistry implements Registry {

    protected static final Logger logger = LoggerFactory.getLogger(ClientInterfaceRegistry.class);

    private final VersionWatchDog versionWatchDog = new VersionWatchDog();
    /**
     * store subscribers
     */
    @Autowired
    protected Subscribers subscribers;
    /**
     * store watchers
     */
    @Autowired
    protected Watchers watchers;
    /**
     * store publishers
     */
    @Autowired
    protected Publishers publishers;
    @Autowired
    protected Config config;
    @Autowired
    protected PushSwitchService pushSwitchService;
    @Autowired
    protected DataExchanger boltDataExchanger;
    @Autowired
    protected RegistryStrategy registryStrategy;
    @Autowired
    protected WrapperInterceptorManager wrapperInterceptorManager;
    @Autowired
    protected WriteDataAcceptor writeDataAcceptor;
    @Autowired
    protected FirePushService firePushService;
    @Autowired
    protected ConfigProvideDataWatcher configProvideDataWatcher;
    @Autowired
    private RegistryScanCallable registryScanCallable;
    @Autowired
    private MetadataCacheRegistry metadataCacheRegistry;
    @Autowired
    private ExecutorManager executorManager;

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread("SessionVerWatchDog", versionWatchDog).start();
        ConcurrentUtils.createDaemonThread("SessionClientWatchDog", new ClientWatchDog()).start();
    }

    @Override
    public void register(Channel channel, StoreData storeData) {

        WrapperInvocation<RegisterInvokeData, Boolean> wrapperInvocation = new WrapperInvocation(new Wrapper<RegisterInvokeData, Boolean>() {
            @Override
            public Boolean call() {

                switch (storeData.getDataType()) {
                case publisher:
                    Publisher publisher = (Publisher) storeData;
                    publisher.setClientInterfaceId(ServerEnv.PROCESS_ID);
                    if (!publishers.add(publisher)) {
                        break;
                    }
                    // All write operations to DataServer (pub/unPub/clientoff/renew/snapshot)
                    // are handed over to WriteDataAcceptor
                    writeDataAcceptor.accept(new PublisherWriteDataRequest(publisher, WriteDataRequest.WriteDataRequestType.publisher));

                    registryStrategy.afterPublisherRegister(publisher);
                    break;
                case subscriber:
                    Subscriber subscriber = (Subscriber) storeData;

                    if (!subscribers.add(subscriber)) {
                        break;
                    }

                    registryStrategy.afterSubscriberRegister(subscriber);
                    break;
                case watcher:
                    Watcher watcher = (Watcher) storeData;

                    if (!watchers.add(watcher)) {
                        break;
                    }

                    registryStrategy.afterWatcherRegister(watcher);
                    break;
                default:
                    break;
                }
                return null;
            }

            @Override
            public Supplier<RegisterInvokeData> getParameterSupplier() {
                return () -> new RegisterInvokeData(storeData, channel);
            }
        }, wrapperInterceptorManager.getInterceptorChain());

        try {
            wrapperInvocation.proceed();
        } catch (RequestChannelClosedException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException("Proceed register error!", e);
        }
    }

    @Override
    public void unRegister(StoreData<String> storeData) {

        switch (storeData.getDataType()) {
        case publisher:
            Publisher publisher = (Publisher) storeData;
            publisher.setClientInterfaceId(ServerEnv.PROCESS_ID);
            // no need to check whether the pub exist, make sure the unpub send to data
            publishers.deleteById(publisher.getDataInfoId(), storeData.getId());
            // All write operations to DataServer (pub/unPub/clientoff)
            // are handed over to WriteDataAcceptor
            writeDataAcceptor.accept(new PublisherWriteDataRequest(publisher, WriteDataRequest.WriteDataRequestType.un_publisher));

            registryStrategy.afterPublisherUnRegister(publisher);
            break;

        case subscriber:
            Subscriber subscriber = (Subscriber) storeData;
            if (subscribers.deleteById(subscriber.getDataInfoId(), storeData.getId()) == null) {
                break;
            }
            registryStrategy.afterSubscriberUnRegister(subscriber);
            break;

        case watcher:
            Watcher watcher = (Watcher) storeData;

            if (watchers.deleteById(watcher.getDataInfoId(), watcher.getId()) == null) {
                break;
            }
            registryStrategy.afterWatcherUnRegister(watcher);
            break;
        default:
            break;
        }
    }

    @Override
    public void clean(List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIdList) {
        disableConnect(clientToClientInterfaceServerNodeIdList, true, false);
    }

    @Override
    public void clientOff(List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIdList) {
        ClientManagerMetric.CLIENT_OFF_COUNTER.inc(clientToClientInterfaceServerNodeIdList.size());
        disableConnect(clientToClientInterfaceServerNodeIdList, false, true);
    }

    @Override
    public void blacklist(List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIdList) {
        disableConnect(clientToClientInterfaceServerNodeIdList, true, true);
    }

    private void disableConnect(List<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds, boolean removeSubAndWat, boolean checkSub) {
        Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIdSet = Collections.unmodifiableSet(new HashSet<>(clientToClientInterfaceServerNodeIds));
        disableConnect(clientToClientInterfaceServerNodeIdSet, removeSubAndWat, checkSub, Collections.emptyMap());
    }

    private void disableConnect(Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIdSet, boolean removeSubAndWat, boolean checkSub, Map<ClientToClientInterfaceServerNodeId, Long> connectIdVersions) {
        if (CollectionUtils.isEmpty(clientToClientInterfaceServerNodeIdSet)) {
            return;
        }
        logger.info("disable connectId={}, removeSubAndWat={}, checkSub={}, {}", clientToClientInterfaceServerNodeIdSet.size(), removeSubAndWat, checkSub, clientToClientInterfaceServerNodeIdSet);

        final String dataCenter = getDataCenterWhenPushEmpty();

        if (checkSub) {
            Map<ClientToClientInterfaceServerNodeId, Map<String, Subscriber>> map = subscribers.queryByConnectIds(clientToClientInterfaceServerNodeIdSet);
            for (Entry<ClientToClientInterfaceServerNodeId, Map<String, Subscriber>> entry : map.entrySet()) {
                int subEmptyCount = 0;
                for (Subscriber subscriber : entry.getValue().values()) {
                    if (isPushEmpty(subscriber)) {
                        Long clientOffVersion = connectIdVersions.get(entry.getKey());
                        if (clientOffVersion != null && clientOffVersion < subscriber.getRegisterTimestamp()) {
                            logger.error("[ClientOffVersionError]subEmpty,{},{},{}, clientOffVersion={} is smaller than subRegisterTimestamp={}", subscriber.getDataInfoId(), dataCenter, entry.getKey(), clientOffVersion, subscriber.getRegisterTimestamp());
                            continue;
                        }

                        subEmptyCount++;
                        firePushService.fireOnPushEmpty(subscriber, dataCenter);
                        logger.info("subEmpty,{},{},{}", subscriber.getDataInfoId(), dataCenter, entry.getKey());
                    }
                }
                logger.info("connectId={}, subEmpty={}", entry.getKey(), subEmptyCount);
            }
        }

        Map<ClientToClientInterfaceServerNodeId, List<Publisher>> pubMap = removeFromSession(clientToClientInterfaceServerNodeIdSet, removeSubAndWat);
        for (Entry<ClientToClientInterfaceServerNodeId, List<Publisher>> entry : pubMap.entrySet()) {
            clientOffToDataNode(entry.getKey(), entry.getValue());
            logger.info("connectId={}, pubRemove={}", entry.getKey(), entry.getValue().size());
        }
    }

    public boolean isPushEmpty(Subscriber subscriber) {
        // mostly, do not need to push empty
        return false;
    }

    private Map<ClientToClientInterfaceServerNodeId, List<Publisher>> removeFromSession(Set<ClientToClientInterfaceServerNodeId> clientToClientInterfaceServerNodeIds, boolean removeSubAndWat) {
        Map<ClientToClientInterfaceServerNodeId, Map<String, Publisher>> publisherMap = publishers.deleteByClientToClientInterfaceServerNodeIdSet(clientToClientInterfaceServerNodeIds);

        if (removeSubAndWat) {
            subscribers.deleteByClientToClientInterfaceServerNodeIdSet(clientToClientInterfaceServerNodeIds);
            watchers.deleteByClientToClientInterfaceServerNodeIdSet(clientToClientInterfaceServerNodeIds);
        }

        Map<ClientToClientInterfaceServerNodeId, List<Publisher>> map = new HashMap<>();
        for (Entry<ClientToClientInterfaceServerNodeId, Map<String, Publisher>> entry : publisherMap.entrySet()) {
            map.put(entry.getKey(), new ArrayList<>(entry.getValue().values()));
        }
        return map;
    }

    private void clientOffToDataNode(ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId, List<Publisher> clientOffPublishers) {
        if (CollectionUtils.isEmpty(clientOffPublishers)) {
            return;
        }
        writeDataAcceptor.accept(new ClientOffWriteDataRequest(clientToClientInterfaceServerNodeId, clientOffPublishers));
    }

    private void scanSubscribers(long round) {

        Set<String> dataCenters = new LinkedHashSet<>();

        dataCenters.add(config.getDataCenter());
        dataCenters.addAll(metadataCacheRegistry.getPushEnableDataCenters());

        final long start = System.currentTimeMillis();
        SelectSubscriber selectSubscriber = subscribers.selectSubscribers(dataCenters);
        logger.info("[select]round={}, regMultiSize={}, span={}", round, selectSubscriber.toRegisterMulti.size(), System.currentTimeMillis() - start);

        regMulti(round, selectSubscriber.toRegisterMulti);

        Map<String, Future<Boolean>> map = new HashMap<>(dataCenters.size());
        for (String dataCenter : dataCenters) {
            Future<Boolean> future = executorManager.getScanExecutor().submit(() -> {
                try {
                    Map<String, DatumVersion> stringDatumVersionMap = selectSubscriber.versions.get(dataCenter);
                    List<Subscriber> pushEmpty = selectSubscriber.toPushEmpty.get(dataCenter);
                    logger.info("[scan]dataCenter={}, round={}, interestSize={}, pushEmptySize={}", dataCenter, round, stringDatumVersionMap.size(), pushEmpty.size(), System.currentTimeMillis() - start);
                    registryScanCallable.scanVersions(round, dataCenter, stringDatumVersionMap, callableInfo -> {
                        if (subscribers.checkInterestVersion(callableInfo.getDataCenter(), callableInfo.getDataInfoId(), callableInfo.getDatumVersion().getValue()).interested) {
                            TriggerPushContext triggerPushContext = new TriggerPushContext(callableInfo.getDataCenter(), callableInfo.getDatumVersion().getValue(), callableInfo.getLeader(), callableInfo.getCurrentTs());
                            firePushService.fireOnChange(callableInfo.getDataInfoId(), triggerPushContext);
                            logger.info("[fetchSlotVerNotify]round={},{},{},{},{}", callableInfo.getRound(), callableInfo.getDatumVersion(), callableInfo.getDataInfoId(), callableInfo.getDataCenter(), callableInfo.getDatumVersion().getValue());
                        }
                    });
                    handlePushEmptySubscribers(dataCenter, pushEmpty);

                    return true;
                } catch (Throwable th) {
                    logger.error("failed to scan version, dataCenter:{}, round:{}", dataCenter, round, th);
                    return false;
                }
            });

            map.put(dataCenter, future);
        }

        for (Entry<String, Future<Boolean>> entry : map.entrySet()) {
            try {
                entry.getValue().get(config.getScanTimeoutMills(), TimeUnit.MILLISECONDS);
            } catch (Throwable th) {
                // return when any datacenter scan timeout
                logger.error("scan version timeout, dataCenter:{}, round:{}", entry.getKey(), round, th);
                return;
            }
        }
    }

    private void regMulti(long round, List<Subscriber> toRegisterMulti) {
        for (Subscriber subscriber : toRegisterMulti) {
            try {
                firePushService.fireOnRegister(subscriber);
            } catch (Throwable e) {
                logger.error("failed to scan subscribers, round:{}, {}", round, subscriber.shortDesc(), e);
            }
        }
    }

    public String getDataCenterWhenPushEmpty() {
        return config.getDataCenter();
    }

    private void handlePushEmptySubscribers(String dataCenter, List<Subscriber> pushEmptySubscribers) {
        for (Subscriber subscriber : pushEmptySubscribers) {
            try {
                if (subscriber.needPushEmpty(dataCenter)) {
                    firePushService.fireOnPushEmpty(subscriber, dataCenter);
                }
            } catch (Throwable e) {
                logger.error("failed to scan subscribers, {}", subscriber.shortDesc(), e);
            }
        }
    }

    public void cleanClientConnect() {
        Server server = boltDataExchanger.getServer(config.getServerPort());
        if (server == null) {
            logger.warn("server not init when clean connect: {}", config.getServerPort());
            return;
        }

        Set<ClientToClientInterfaceServerNodeId> set = new HashSet<>(1024 * 8);
        set.addAll(publishers.getConnectIds());
        set.addAll(subscribers.getConnectIds());
        set.addAll(watchers.getConnectIds());

        List<ClientToClientInterfaceServerNodeId> list = new ArrayList<>(64);
        for (ClientToClientInterfaceServerNodeId clientToClientInterfaceServerNodeId : set) {
            Channel channel = server.getChannel(new Url(clientToClientInterfaceServerNodeId.getClientIp(), clientToClientInterfaceServerNodeId.getClientPort()));
            if (channel == null) {
                list.add(clientToClientInterfaceServerNodeId);
            }
        }
        clean(list);
    }

    public static class SelectSubscriber {
        final Map<String /*dataCenter*/, Map<String /*dataInfoId*/, DatumVersion>> versions;

        final Map<String /*dataCenter*/, List<Subscriber>> toPushEmpty;

        final List<Subscriber> toRegisterMulti;

        public SelectSubscriber(Map<String, Map<String, DatumVersion>> versions, Map<String, List<Subscriber>> toPushEmpty, List<Subscriber> toRegisterMulti) {
            this.versions = versions;
            this.toPushEmpty = toPushEmpty;
            this.toRegisterMulti = toRegisterMulti;
        }

    }

    private final class ClientWatchDog extends LoopExecuteTask {

        @Override
        public void _execute_() {
            try {
                cleanClientConnect();
            } catch (Throwable e) {
                logger.error("WatchDog failed to cleanClientConnect", e);
            }
        }

        @Override
        public void _wait_() {
            ConcurrentUtils.sleepUninterruptibly(5, TimeUnit.SECONDS);
        }
    }

    private final class VersionWatchDog extends WakeUpLoopExecuteTask {
        boolean prevStopPushSwitch;
        long scanRound;
        long lastScanTimestamp;

        @Override
        public void _execute_() {
            try {
                final int intervalMillis = config.getScanSubscriberIntervalMillis();
                final boolean stop = !pushSwitchService.canLocalDataCenterPush();
                // could not start scan ver at begin
                // 1. stopPush.val = true default in session.default
                if (stop) {
                    logger.info("[stopPush]");
                    prevStopPushSwitch = true;
                    return;
                }
                final long now = System.currentTimeMillis();
                // abs avoid the clock attack
                if (Math.abs(now - lastScanTimestamp) >= intervalMillis || prevStopPushSwitch) {
                    try {
                        scanSubscribers(scanRound++);
                    } finally {
                        lastScanTimestamp = System.currentTimeMillis();
                    }
                }
                prevStopPushSwitch = false;
            } catch (Throwable e) {
                logger.error("WatchDog failed fetch versions", e);
            }
        }

        @Override
        public int getWaitingMillis() {
            return 1000;
        }
    }
}
